package com.zhaosc.rx.example;

import java.util.List;
import java.util.concurrent.TimeUnit;

import org.junit.Test;

import com.zhaosc.rx.utils.SleepUtils;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.observables.GroupedObservable;

public class TestFilterOperation {

	/**
	 * map():对序列的每一项都用一个函数来变换 Observable发射的数据序列。
	 */
	@Test
	public void testMap() {
		Observable.just("HELLO").map(s -> s.toLowerCase()).map(s -> s + "zhaosc").subscribe(new Consumer<String>() {

			@Override
			public void accept(String t) throws Exception {
				System.out.println("******" + t);
			}
		});
	}

	/**
	 * flatMap()、 concatMap()和 flatMaplterable(): 将 Observable 发射的数据集合变换为
	 * Observables 集合， 然后将这些 Observable 发射的数据平坦化地放进 一 个单独的Observable 中 。
	 */
	@Test
	public void testFlapMap() {
		Observable.just("HELLO_World", "zhaosc").flatMap(new Function<String, ObservableSource<String>>() {

			@Override
			public ObservableSource<String> apply(String t) throws Exception {

				return Observable.fromArray(t.split("_"));
			}
		}).subscribe(new Consumer<String>() {

			@Override
			public void accept(String t) throws Exception {
				System.out.println("******" + t);
			}
		});
	}

	/**
	 * groupBy 操作符将一个 Observable 拆分为一些 Observables 集合， 它们中的每一个都发射原始Observable 的一个子序
	 * 列，
	 */
	@Test
	public void testGroupBy() {
		Observable.range(1, 10).groupBy(new Function<Integer, String>() {

			@Override
			public String apply(Integer t) throws Exception {
				return t % 2 == 0 ? "O" : "J";
			}
		}).subscribe(new Consumer<GroupedObservable<String, Integer>>() {
			@Override
			public void accept(GroupedObservable<String, Integer> t) throws Exception {
				String key = t.getKey();
				t.subscribe(new Consumer<Integer>() {

					@Override
					public void accept(Integer t) throws Exception {
						System.out.println(key + "*****" + t);
					}
				});
			}
		});

	}

	/**
	 * buffer 会定期收集 Observable 数据并放进一个数据包裹，然后发射这些数据包裹，而不是 一次发射一个值
	 * 
	 */
	@Test
	public void testBuffer() {
		Observable.range(1, 10).buffer(3)
				// .buffer(3, 2)
				.subscribe(new Consumer<List<Integer>>() {

					@Override
					public void accept(List<Integer> t) throws Exception {

						t.forEach(new java.util.function.Consumer<Integer>() {
							@Override
							public void accept(Integer t) {
								System.out.print(t + "  ");
							}
						});

					}
				});

	}

	@Test
	public void testWindowns() {
		Observable.range(1, 10).window(3).subscribe(new Consumer<Observable<Integer>>() {
			@Override
			public void accept(Observable<Integer> t) throws Exception {
				System.out.println("---------");
				t.subscribe(new Consumer<Integer>() {

					@Override
					public void accept(Integer t) throws Exception {
						System.out.println("****" + t);
					}
				});
			}
		});
	}

	/**
	 * 只发射第一项或者最后一项
	 */
	@Test
	public void testFirstAndLast() {
		Observable
				// .<String>empty()
				.just("1", "2", "3", "4", "5", "6")
				// .first("first1")
				.last("last").subscribe(new Consumer<String>() {

					@Override
					public void accept(String t) throws Exception {
						System.out.println(t);
					}
				});
	}

	/**
	 * 只发射第一项或者最后一项
	 */
	@Test
	public void testTakeAndLast() {
		Observable
				// .<String>empty()
				.just("1", "2", "3", "4", "5", "6")
				// .first("first1")
				.take(3).skip(2).subscribe(new Consumer<String>() {

					@Override
					public void accept(String t) throws Exception {
						System.out.println(t);
					}
				});
	}

	@Test
	public void testElementAt() {
		Observable
				// .<String>empty()
				.just("1", "2", "3", "4", "5", "6")
				// .first("first1")
				.elementAt(3).subscribe(new Consumer<String>() {

					@Override
					public void accept(String t) throws Exception {
						System.out.println(t);
					}
				});
	}

	@Test
	public void testDistinct() {
		Observable
				// .<String>empty()
				.just("1", "2", "1", "4", "5", "5")
				// .first("first1")
				// .distinct()
//		.distinct(new Function<String, Integer>() {
//
//			@Override
//			public Integer apply(String t) throws Exception {
//				return t.hashCode();
//			}
//		})
				.distinctUntilChanged().subscribe(new Consumer<String>() {

					@Override
					public void accept(String t) throws Exception {
						System.out.println(t);
					}
				});
	}

	@Test
	public void testFilter() {
		Observable
		.just("1", "2","1", "4", "5", "5")
		.filter(t->t.hashCode()%2==0)

		.distinctUntilChanged()
		.subscribe(new Consumer<String>() {

			@Override
			public void accept(String t) throws Exception {
				System.out.println(t);
			}
		});
	}

	/**
	 * debounce 操作符会过滤掉发射速率过快的数据项
	 */
	@Test
	public void testDebounce() {
		Observable.create(new ObservableOnSubscribe<Integer>() {

			@Override
			public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
				for(int i=0;i<10;i++) {
					emitter.onNext(i);
					SleepUtils.sleep(i*100L);
				}
			}
		})
		.debounce(500, TimeUnit.MILLISECONDS)
		
//		.debounce(new Function<Integer, ObservableSource<String>>() {
//
//			@Override
//			public ObservableSource<String> apply(Integer t) throws Exception {
//				return Observable.just(t+"");
//			}
//		})
		.subscribe(new Consumer<Integer>() {

			@Override
			public void accept(Integer t) throws Exception {
				System.out.println("TTTT"+t);
			}
		});
	}
}
